package com.cencat.common.utils;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.Supplier;
import java.util.List;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 并发工具类
 * 提供线程池管理、并发执行、锁操作等功能
 * 
 * @author cencat
 * @since 2024-01-01
 */

public class ConcurrentUtils {

    private static final Logger log = LoggerFactory.getLogger(ConcurrentUtils.class);

    /**
     * 默认线程池
     */
    private static final ThreadPoolExecutor DEFAULT_EXECUTOR;
    
    /**
     * 线程池映射
     */
    private static final Map<String, ThreadPoolExecutor> EXECUTOR_MAP = new ConcurrentHashMap<>();
    
    /**
     * 锁映射
     */
    private static final Map<String, ReentrantLock> LOCK_MAP = new ConcurrentHashMap<>();
    
    /**
     * 读写锁映射
     */
    private static final Map<String, ReentrantReadWriteLock> READ_WRITE_LOCK_MAP = new ConcurrentHashMap<>();

    static {
        // 初始化默认线程池
        int corePoolSize = Runtime.getRuntime().availableProcessors();
        int maximumPoolSize = corePoolSize * 2;
        long keepAliveTime = 60L;
        TimeUnit unit = TimeUnit.SECONDS;
        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(1000);
        ThreadFactory threadFactory = new NamedThreadFactory("cencat-default");
        RejectedExecutionHandler handler = new ThreadPoolExecutor.CallerRunsPolicy();
        
        DEFAULT_EXECUTOR = new ThreadPoolExecutor(
                corePoolSize, maximumPoolSize, keepAliveTime, unit, 
                workQueue, threadFactory, handler);
        
        // 注册JVM关闭钩子
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            log.info("开始关闭线程池...");
            shutdownAllExecutors();
            log.info("线程池关闭完成");
        }));
    }

    private ConcurrentUtils() {
        // 私有构造函数，防止实例化
    }

    /**
     * 获取默认线程池
     * 
     * @return 默认线程池
     */
    public static ThreadPoolExecutor getDefaultExecutor() {
        return DEFAULT_EXECUTOR;
    }

    /**
     * 创建线程池
     * 
     * @param name           线程池名称
     * @param corePoolSize   核心线程数
     * @param maximumPoolSize 最大线程数
     * @param keepAliveTime  线程存活时间
     * @param unit           时间单位
     * @param queueCapacity  队列容量
     * @return 线程池
     */
    public static ThreadPoolExecutor createExecutor(String name, int corePoolSize, 
                                                   int maximumPoolSize, long keepAliveTime, 
                                                   TimeUnit unit, int queueCapacity) {
        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(queueCapacity);
        ThreadFactory threadFactory = new NamedThreadFactory(name);
        RejectedExecutionHandler handler = new ThreadPoolExecutor.CallerRunsPolicy();
        
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                corePoolSize, maximumPoolSize, keepAliveTime, unit, 
                workQueue, threadFactory, handler);
        
        EXECUTOR_MAP.put(name, executor);
        log.info("创建线程池: name={}, corePoolSize={}, maximumPoolSize={}, queueCapacity={}", 
                name, corePoolSize, maximumPoolSize, queueCapacity);
        
        return executor;
    }

    /**
     * 获取线程池
     * 
     * @param name 线程池名称
     * @return 线程池
     */
    public static ThreadPoolExecutor getExecutor(String name) {
        return EXECUTOR_MAP.get(name);
    }

    /**
     * 异步执行任务
     * 
     * @param task 任务
     * @return Future
     */
    public static Future<?> executeAsync(Runnable task) {
        return DEFAULT_EXECUTOR.submit(task);
    }

    /**
     * 异步执行任务（指定线程池）
     * 
     * @param executorName 线程池名称
     * @param task         任务
     * @return Future
     */
    public static Future<?> executeAsync(String executorName, Runnable task) {
        ThreadPoolExecutor executor = getExecutor(executorName);
        if (executor == null) {
            log.warn("线程池不存在，使用默认线程池: executorName={}", executorName);
            executor = DEFAULT_EXECUTOR;
        }
        return executor.submit(task);
    }

    /**
     * 异步执行有返回值的任务
     * 
     * @param task 任务
     * @param <T>  返回值类型
     * @return Future
     */
    public static <T> Future<T> submitAsync(Callable<T> task) {
        return DEFAULT_EXECUTOR.submit(task);
    }

    /**
     * 异步执行有返回值的任务（指定线程池）
     * 
     * @param executorName 线程池名称
     * @param task         任务
     * @param <T>          返回值类型
     * @return Future
     */
    public static <T> Future<T> submitAsync(String executorName, Callable<T> task) {
        ThreadPoolExecutor executor = getExecutor(executorName);
        if (executor == null) {
            log.warn("线程池不存在，使用默认线程池: executorName={}", executorName);
            executor = DEFAULT_EXECUTOR;
        }
        return executor.submit(task);
    }

    /**
     * 并行执行多个任务
     * 
     * @param tasks 任务列表
     * @return Future列表
     */
    public static List<Future<?>> executeParallel(List<Runnable> tasks) {
        if (tasks == null || tasks.isEmpty()) {
            return new ArrayList<>();
        }
        
        List<Future<?>> futures = new ArrayList<>();
        for (Runnable task : tasks) {
            futures.add(DEFAULT_EXECUTOR.submit(task));
        }
        return futures;
    }

    /**
     * 并行执行多个有返回值的任务
     * 
     * @param tasks 任务列表
     * @param <T>   返回值类型
     * @return Future列表
     */
    public static <T> List<Future<T>> submitParallel(List<Callable<T>> tasks) {
        if (tasks == null || tasks.isEmpty()) {
            return new ArrayList<>();
        }
        
        List<Future<T>> futures = new ArrayList<>();
        for (Callable<T> task : tasks) {
            futures.add(DEFAULT_EXECUTOR.submit(task));
        }
        return futures;
    }

    /**
     * 等待所有任务完成
     * 
     * @param futures Future列表
     * @param timeout 超时时间
     * @param unit    时间单位
     * @return 是否全部完成
     */
    public static boolean waitForAll(List<Future<?>> futures, long timeout, TimeUnit unit) {
        if (futures == null || futures.isEmpty()) {
            return true;
        }
        
        long deadline = System.nanoTime() + unit.toNanos(timeout);
        for (Future<?> future : futures) {
            try {
                long remaining = deadline - System.nanoTime();
                if (remaining <= 0) {
                    return false;
                }
                future.get(remaining, TimeUnit.NANOSECONDS);
            } catch (Exception e) {
                log.error("等待任务完成失败: error={}", e.getMessage());
                return false;
            }
        }
        return true;
    }

    /**
     * 获取所有任务结果
     * 
     * @param futures Future列表
     * @param timeout 超时时间
     * @param unit    时间单位
     * @param <T>     返回值类型
     * @return 结果列表
     */
    public static <T> List<T> getAllResults(List<Future<T>> futures, long timeout, TimeUnit unit) {
        if (futures == null || futures.isEmpty()) {
            return new ArrayList<>();
        }
        
        List<T> results = new ArrayList<>();
        long deadline = System.nanoTime() + unit.toNanos(timeout);
        
        for (Future<T> future : futures) {
            try {
                long remaining = deadline - System.nanoTime();
                if (remaining <= 0) {
                    log.warn("获取任务结果超时");
                    break;
                }
                T result = future.get(remaining, TimeUnit.NANOSECONDS);
                results.add(result);
            } catch (Exception e) {
                log.error("获取任务结果失败: error={}", e.getMessage());
            }
        }
        
        return results;
    }

    /**
     * 延迟执行任务
     * 
     * @param task  任务
     * @param delay 延迟时间
     * @param unit  时间单位
     * @return ScheduledFuture
     */
    public static ScheduledFuture<?> scheduleDelay(Runnable task, long delay, TimeUnit unit) {
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1, 
                new NamedThreadFactory("cencat-scheduler"));
        return scheduler.schedule(task, delay, unit);
    }

    /**
     * 定时执行任务
     * 
     * @param task         任务
     * @param initialDelay 初始延迟
     * @param period       执行周期
     * @param unit         时间单位
     * @return ScheduledFuture
     */
    public static ScheduledFuture<?> scheduleAtFixedRate(Runnable task, long initialDelay, 
                                                        long period, TimeUnit unit) {
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1, 
                new NamedThreadFactory("cencat-scheduler"));
        return scheduler.scheduleAtFixedRate(task, initialDelay, period, unit);
    }

    /**
     * 获取锁
     * 
     * @param lockName 锁名称
     * @return 锁
     */
    public static ReentrantLock getLock(String lockName) {
        return LOCK_MAP.computeIfAbsent(lockName, k -> new ReentrantLock());
    }

    /**
     * 获取读写锁
     * 
     * @param lockName 锁名称
     * @return 读写锁
     */
    public static ReentrantReadWriteLock getReadWriteLock(String lockName) {
        return READ_WRITE_LOCK_MAP.computeIfAbsent(lockName, k -> new ReentrantReadWriteLock());
    }

    /**
     * 使用锁执行任务
     * 
     * @param lockName 锁名称
     * @param task     任务
     */
    public static void executeWithLock(String lockName, Runnable task) {
        ReentrantLock lock = getLock(lockName);
        lock.lock();
        try {
            task.run();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 使用锁执行有返回值的任务
     * 
     * @param lockName 锁名称
     * @param supplier 任务
     * @param <T>      返回值类型
     * @return 任务结果
     */
    public static <T> T executeWithLock(String lockName, Supplier<T> supplier) {
        ReentrantLock lock = getLock(lockName);
        lock.lock();
        try {
            return supplier.get();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 使用读锁执行任务
     * 
     * @param lockName 锁名称
     * @param supplier 任务
     * @param <T>      返回值类型
     * @return 任务结果
     */
    public static <T> T executeWithReadLock(String lockName, Supplier<T> supplier) {
        ReentrantReadWriteLock readWriteLock = getReadWriteLock(lockName);
        ReentrantReadWriteLock.ReadLock readLock = readWriteLock.readLock();
        readLock.lock();
        try {
            return supplier.get();
        } finally {
            readLock.unlock();
        }
    }

    /**
     * 使用写锁执行任务
     * 
     * @param lockName 锁名称
     * @param task     任务
     */
    public static void executeWithWriteLock(String lockName, Runnable task) {
        ReentrantReadWriteLock readWriteLock = getReadWriteLock(lockName);
        ReentrantReadWriteLock.WriteLock writeLock = readWriteLock.writeLock();
        writeLock.lock();
        try {
            task.run();
        } finally {
            writeLock.unlock();
        }
    }

    /**
     * 使用写锁执行有返回值的任务
     * 
     * @param lockName 锁名称
     * @param supplier 任务
     * @param <T>      返回值类型
     * @return 任务结果
     */
    public static <T> T executeWithWriteLock(String lockName, Supplier<T> supplier) {
        ReentrantReadWriteLock readWriteLock = getReadWriteLock(lockName);
        ReentrantReadWriteLock.WriteLock writeLock = readWriteLock.writeLock();
        writeLock.lock();
        try {
            return supplier.get();
        } finally {
            writeLock.unlock();
        }
    }

    /**
     * 尝试获取锁并执行任务
     * 
     * @param lockName 锁名称
     * @param task     任务
     * @param timeout  超时时间
     * @param unit     时间单位
     * @return 是否执行成功
     */
    public static boolean tryExecuteWithLock(String lockName, Runnable task, 
                                           long timeout, TimeUnit unit) {
        ReentrantLock lock = getLock(lockName);
        try {
            if (lock.tryLock(timeout, unit)) {
                try {
                    task.run();
                    return true;
                } finally {
                    lock.unlock();
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("获取锁被中断: lockName={}, error={}", lockName, e.getMessage());
        }
        return false;
    }

    /**
     * 批量处理数据
     * 
     * @param data      数据集合
     * @param batchSize 批次大小
     * @param processor 处理器
     * @param <T>       数据类型
     */
    public static <T> void processBatch(Collection<T> data, int batchSize, 
                                      BatchProcessor<T> processor) {
        if (data == null || data.isEmpty() || processor == null) {
            return;
        }
        
        List<T> dataList = new ArrayList<>(data);
        int totalSize = dataList.size();
        int batchCount = (totalSize + batchSize - 1) / batchSize;
        
        List<Future<?>> futures = new ArrayList<>();
        
        for (int i = 0; i < batchCount; i++) {
            int start = i * batchSize;
            int end = Math.min(start + batchSize, totalSize);
            List<T> batch = dataList.subList(start, end);
            
            Future<?> future = DEFAULT_EXECUTOR.submit(() -> {
                try {
                    processor.process(batch);
                } catch (Exception e) {
                    log.error("批量处理失败: batch={}, error={}", batch.size(), e.getMessage());
                }
            });
            
            futures.add(future);
        }
        
        // 等待所有批次处理完成
        waitForAll(futures, 30, TimeUnit.SECONDS);
    }

    /**
     * 并行处理数据
     * 
     * @param data      数据集合
     * @param processor 处理器
     * @param <T>       数据类型
     */
    public static <T> void processParallel(Collection<T> data, ItemProcessor<T> processor) {
        if (data == null || data.isEmpty() || processor == null) {
            return;
        }
        
        List<Future<?>> futures = new ArrayList<>();
        
        for (T item : data) {
            Future<?> future = DEFAULT_EXECUTOR.submit(() -> {
                try {
                    processor.process(item);
                } catch (Exception e) {
                    log.error("并行处理失败: item={}, error={}", item, e.getMessage());
                }
            });
            
            futures.add(future);
        }
        
        // 等待所有任务处理完成
        waitForAll(futures, 30, TimeUnit.SECONDS);
    }

    /**
     * 获取线程池状态
     * 
     * @param executorName 线程池名称
     * @return 线程池状态
     */
    public static ExecutorStatus getExecutorStatus(String executorName) {
        ThreadPoolExecutor executor = getExecutor(executorName);
        if (executor == null) {
            executor = DEFAULT_EXECUTOR;
        }
        
        return new ExecutorStatus(
                executor.getCorePoolSize(),
                executor.getMaximumPoolSize(),
                executor.getPoolSize(),
                executor.getActiveCount(),
                executor.getQueue().size(),
                executor.getCompletedTaskCount(),
                executor.getTaskCount()
        );
    }

    /**
     * 关闭线程池
     * 
     * @param executorName 线程池名称
     */
    public static void shutdownExecutor(String executorName) {
        ThreadPoolExecutor executor = EXECUTOR_MAP.remove(executorName);
        if (executor != null) {
            shutdownExecutor(executor, executorName);
        }
    }

    /**
     * 关闭所有线程池
     */
    public static void shutdownAllExecutors() {
        // 关闭默认线程池
        shutdownExecutor(DEFAULT_EXECUTOR, "default");
        
        // 关闭所有自定义线程池
        EXECUTOR_MAP.forEach((name, executor) -> {
            shutdownExecutor(executor, name);
        });
        
        EXECUTOR_MAP.clear();
    }

    /**
     * 关闭线程池
     * 
     * @param executor 线程池
     * @param name     线程池名称
     */
    private static void shutdownExecutor(ThreadPoolExecutor executor, String name) {
        if (executor == null || executor.isShutdown()) {
            return;
        }
        
        try {
            log.info("开始关闭线程池: name={}", name);
            executor.shutdown();
            
            if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
                log.warn("线程池未能在30秒内关闭，强制关闭: name={}", name);
                executor.shutdownNow();
                
                if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {
                    log.error("线程池强制关闭失败: name={}", name);
                }
            }
            
            log.info("线程池关闭完成: name={}", name);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("关闭线程池被中断: name={}, error={}", name, e.getMessage());
            executor.shutdownNow();
        }
    }

    /**
     * 命名线程工厂
     */
    public static class NamedThreadFactory implements ThreadFactory {
        private final String namePrefix;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final boolean daemon;

        public NamedThreadFactory(String namePrefix) {
            this(namePrefix, false);
        }

        public NamedThreadFactory(String namePrefix, boolean daemon) {
            this.namePrefix = namePrefix;
            this.daemon = daemon;
            // 直接使用当前线程的线程组，避免使用已过时的SecurityManager
            this.group = Thread.currentThread().getThreadGroup();
        }

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r, namePrefix + "-thread-" + threadNumber.getAndIncrement(), 0);
            t.setDaemon(daemon);
            if (t.getPriority() != Thread.NORM_PRIORITY) {
                t.setPriority(Thread.NORM_PRIORITY);
            }
            return t;
        }
    }

    /**
     * 批量处理器接口
     * 
     * @param <T> 数据类型
     */
    @FunctionalInterface
    public interface BatchProcessor<T> {
        /**
         * 处理批量数据
         * 
         * @param batch 批量数据
         */
        void process(List<T> batch);
    }

    /**
     * 单项处理器接口
     * 
     * @param <T> 数据类型
     */
    @FunctionalInterface
    public interface ItemProcessor<T> {
        /**
         * 处理单项数据
         * 
         * @param item 单项数据
         */
        void process(T item);
    }

    /**
     * 线程池状态类
     */
    public static class ExecutorStatus {
        private final int corePoolSize;
        private final int maximumPoolSize;
        private final int poolSize;
        private final int activeCount;
        private final int queueSize;
        private final long completedTaskCount;
        private final long taskCount;

        public ExecutorStatus(int corePoolSize, int maximumPoolSize, int poolSize, 
                            int activeCount, int queueSize, long completedTaskCount, long taskCount) {
            this.corePoolSize = corePoolSize;
            this.maximumPoolSize = maximumPoolSize;
            this.poolSize = poolSize;
            this.activeCount = activeCount;
            this.queueSize = queueSize;
            this.completedTaskCount = completedTaskCount;
            this.taskCount = taskCount;
        }

        // Getter方法
        public int getCorePoolSize() { return corePoolSize; }
        public int getMaximumPoolSize() { return maximumPoolSize; }
        public int getPoolSize() { return poolSize; }
        public int getActiveCount() { return activeCount; }
        public int getQueueSize() { return queueSize; }
        public long getCompletedTaskCount() { return completedTaskCount; }
        public long getTaskCount() { return taskCount; }

        @Override
        public String toString() {
            return String.format(
                    "ExecutorStatus{corePoolSize=%d, maximumPoolSize=%d, poolSize=%d, " +
                    "activeCount=%d, queueSize=%d, completedTaskCount=%d, taskCount=%d}",
                    corePoolSize, maximumPoolSize, poolSize, activeCount, 
                    queueSize, completedTaskCount, taskCount);
        }
    }

    /**
     * 并发工具常量类
     */
    public static class ConcurrentConstants {
        
        // 默认线程池配置
        public static final int DEFAULT_CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors();
        public static final int DEFAULT_MAXIMUM_POOL_SIZE = DEFAULT_CORE_POOL_SIZE * 2;
        public static final long DEFAULT_KEEP_ALIVE_TIME = 60L;
        public static final TimeUnit DEFAULT_TIME_UNIT = TimeUnit.SECONDS;
        public static final int DEFAULT_QUEUE_CAPACITY = 1000;
        
        // 默认超时时间
        public static final long DEFAULT_TIMEOUT = 30L;
        public static final TimeUnit DEFAULT_TIMEOUT_UNIT = TimeUnit.SECONDS;
        
        // 默认批次大小
        public static final int DEFAULT_BATCH_SIZE = 100;
        
        // 线程池名称
        public static final String DEFAULT_EXECUTOR_NAME = "cencat-default";
        public static final String SCHEDULER_EXECUTOR_NAME = "cencat-scheduler";
        
        private ConcurrentConstants() {
            // 私有构造函数，防止实例化
        }
    }
}